package com.CST.service.impl.ys;

import com.CST.common.BaseEntity;
import com.CST.common.api.ResultObject;
import com.CST.common.constant.Constants;
import com.CST.dto.ys.YsComplaintDTO;
import com.CST.entity.ys.YsComplaint;
import com.CST.mapper.cst.ys.YsComplaintMapper;
import com.CST.service.ys.YsComplaintService;
import com.CST.util.HttpUtil;
import com.CST.websocket.WebSocketServer;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.Builder;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.DefaultManagedAwareThreadFactory;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * <p>
 * 服务实现类
 * </p>
 *
 * @author qtx
 * @since 2023-05-15
 */
@Slf4j
@Service
public class YsComplaintServiceImpl extends ServiceImpl<YsComplaintMapper, YsComplaint> implements YsComplaintService {

    private static final ConcurrentHashMap<String, SendMsg> msg = new ConcurrentHashMap<>();

    private final ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 10, 1, TimeUnit.MINUTES,
            new ArrayBlockingQueue<>(20),
            new DefaultManagedAwareThreadFactory(), new ThreadPoolExecutor.AbortPolicy());

    @Override
    public Page<YsComplaint> listYsComplaintPage(YsComplaintDTO dto) {
        return page(dto.getPage(), Wrappers.lambdaQuery(YsComplaint.class)
                .eq(YsComplaint::getComplaintUser, HttpUtil.getRequestHeaderInfo(Constants.USER_NAME))
                .orderByAsc(YsComplaint::getReadMsg));
    }

    @Override
    public List<YsComplaint> online() {
        return list(Wrappers.lambdaQuery(YsComplaint.class)
                .eq(BaseEntity::getDeleteFlag, Boolean.FALSE)
                .eq(YsComplaint::getReadMsg, Boolean.FALSE)
                .eq(YsComplaint::getComplaintUser, HttpUtil.getRequestHeaderInfo(Constants.USER_NAME)));
    }

    @Override
    public boolean updateYsComplaint(Long id) {
        return update(Wrappers.lambdaUpdate(YsComplaint.class)
                .eq(YsComplaint::getComplaintUser, HttpUtil.getRequestHeaderInfo(Constants.USER_NAME))
                .eq(id != null, BaseEntity::getId, id)
                .set(YsComplaint::getReadMsg, Boolean.TRUE));
    }

    @Override
    public boolean saveEntity(YsComplaint entity) {
        return save(entity);
    }

    /**
     * 循环发送
     */
    @PostConstruct
    private void sendMassage() {
        executor.execute(() -> {
            int i = 0;
            while (true) {
                if (!WebSocketServer.webSocketClientMap.isEmpty()) {
                    log.info("开始检索消息。");
                    List<YsComplaint> list = list(Wrappers.lambdaQuery(YsComplaint.class)
                            .eq(BaseEntity::getDeleteFlag, Boolean.FALSE)
                            .eq(YsComplaint::getReadMsg, Boolean.FALSE));
                    Map<String, List<YsComplaint>> map = list.stream()
                            .collect(Collectors.groupingBy(YsComplaint::getComplaintUser));
                    WebSocketServer.webSocketClientMap.forEach((k, v) -> {
                        int count = Optional.ofNullable(map.get(k))
                                .orElse(new ArrayList<>())
                                .size();
                        SendMsg sendMsg = SendMsg.builder()
                                .count(count)
                                .time(LocalDateTime.now())
                                .build();
                        log.info("user:{},count:{}", k, count);
                        String jsonString = JSONObject.toJSONString(ResultObject.success(count));
                        if (msg.containsKey(k) && !(msg.get(k).count == count)) {
                            WebSocketServer.sendMessage(k, jsonString);
                            msg.put(k, sendMsg);
                        }
                        if (!msg.containsKey(k)) {
                            WebSocketServer.sendMessage(k, jsonString);
                            msg.put(k, sendMsg);
                        }
                    });
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    i++;
                    if (i > 20) {
                        clear(Boolean.FALSE);
                        i = 0;
                    }
                } else {
                    if (!msg.isEmpty()) {
                        clear(Boolean.TRUE);
                        i = 0;
                        try {
                            TimeUnit.SECONDS.sleep(1);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                    }
                }
            }
        });
    }

    private void clear(boolean flag) {
        log.info("清理开始。");
        List<String> list = new ArrayList<>();
        msg.forEach((k, v) -> {
            if (flag) {
                list.add(k);
            } else if (v.time.plusMinutes(5)
                    .isBefore(LocalDateTime.now())) {
                list.add(k);
            }
        });
        log.info("清理个数：{}", list.size());
        list.forEach(msg::remove);
    }

    public void instantlySendMsg() {
        String headerInfo = HttpUtil.getRequestHeaderInfo(Constants.USER_NAME);
        String s = Optional.ofNullable(headerInfo)
                .orElseThrow(() -> new RuntimeException("消息来源异常。"));
        int count = count(Wrappers.lambdaQuery(YsComplaint.class)
                .eq(BaseEntity::getDeleteFlag, Boolean.FALSE)
                .eq(YsComplaint::getReadMsg, Boolean.FALSE)
                .eq(YsComplaint::getComplaintUser, headerInfo));
        WebSocketServer.sendMessage(s, JSONObject.toJSONString(ResultObject.success(count)));
    }

    @Data
    @Builder
    private static class SendMsg {
        Integer count;

        LocalDateTime time;
    }
}
